package com.slipper.collect;

import com.slipper.PaasContext;
import com.slipper.codeless.resolve.SqlBuilderFactory;
import com.slipper.codeless.resolve.query.Condition;
import com.slipper.codeless.resolve.query.Query;
import com.slipper.codeless.resolve.query.QueryField;
import com.slipper.datasource.DatasourceUtil;
import com.slipper.exception.CommonException;
import com.slipper.system.codeless.bo.DatasourceBO;
import com.slipper.system.codeless.service.DatasourceService;
import com.slipper.system.collect.bo.CollectColumnBO;
import com.slipper.system.collect.bo.CollectTableBO;
import com.slipper.system.collect.bo.CollectTableLogBO;
import com.slipper.system.collect.bo.CollectTaskLogBO;
import com.slipper.system.collect.service.CollectTableLogService;
import com.slipper.system.collect.service.CollectTableService;
import com.slipper.util.Tool;
import com.slipper.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class CollectTable implements Runnable {
    private final Logger logger = LoggerFactory.getLogger(CollectTable.class);

    private final CollectTaskLogBO collectTaskLog;
    private final CollectTableBO collectTable;
    private int addCount = 0;
    private int updateCount = 0;
    private boolean shutdown = false;
    private boolean success = false;

    public CollectTable(CollectTaskLogBO collectTaskLog, CollectTableBO collectTable) {
        this.collectTaskLog = collectTaskLog;
        this.collectTable = collectTable;
    }

    @Override
    public void run() {
        if (shutdown) return;

        CollectTableLogBO collectTableLog = createLog(collectTaskLog.getId(), collectTable);
        CollectTableLogService collectTableLogService = PaasContext.getBean(CollectTableLogService.class);
        collectTableLogService.insert(collectTableLog);

        try {
            if (CollectTableBO.SYNC_TYPE_ADD.equals(collectTable.getSyncType())) {
                collectTable.setIncrementStartTime(collectTable.getIncrementEndTime());
                collectTable.setIncrementEndTime(System.currentTimeMillis());
            }

            DatasourceService datasourceService = PaasContext.getBean(DatasourceService.class);
            DatasourceBO originDatasource = datasourceService.queryBy(collectTaskLog.getOriginDatasourceId());
            DatasourceBO targetDatasource = datasourceService.queryBy(collectTaskLog.getTargetDatasourceId());

            // 查询源表结果数
            long originTotal = DatasourceUtil.queryTotal(
                    originDatasource,
                    createQueryCountMap(originDatasource, collectTable));
            collectTableLog.setOriginDataCount(originTotal);
            collectTableLogService.update(collectTableLog);

            // 前置处理
            before(targetDatasource, collectTable.getTargetTable());

            List<String> columnList = getColumnList(collectTable);

            int pageNumber = 1;
            int pageSize = collectTable.getReadCount() == null ? 3000 : collectTable.getReadCount();
            boolean hasNext = true;
            int targetTotal = 0;
            Map<String, Object> queryMap = createQueryListMap(originDatasource, collectTable);
            while(!shutdown && hasNext) {
                List<Map<String, Object>> dataList = DatasourceUtil.queryPage(
                        originDatasource,
                        pageNumber,
                        pageSize,
                        queryMap
                ).getList();
                targetTotal += dataList.size();

                convertDataList(collectTable, dataList);
                saveDataList(targetDatasource, collectTable, collectTableLog, dataList, columnList);
                collectTableLogService.update(collectTableLog);

                if (targetTotal > originTotal || dataList.size() < pageSize || dataList.isEmpty()) {
                    hasNext = false;
                }
                pageNumber++;
            }

            if (CollectTableBO.SYNC_TYPE_ADD.equals(collectTable.getSyncType())) {
                PaasContext.getBean(CollectTableService.class).save(collectTable);
            }

            success = true;
            collectTableLog.setExecuteStatus("执行成功");
        } catch (Exception e) {
            success = false;
            logger.error("任务执行失败：", e);
            collectTableLog.setExecuteStatus("执行失败");
            collectTableLog.setErrorMessage(e.getMessage());
        } finally {
            collectTableLog.setEndTime(System.currentTimeMillis());
            collectTableLog.setExecuteTime(collectTableLog.getEndTime() - collectTableLog.getStartTime());
            collectTableLogService.update(collectTableLog);
        }
    }

    private void before(DatasourceBO datasource, String tableName) {
        if (CollectTableBO.WRITE_MODE_DW.equals(collectTable.getWriteMode())) {
            DatasourceUtil.deleteAll(datasource, tableName);
        }
    }

    private List<String> getColumnList(CollectTableBO collectTable) {
        List<String> columnList = new ArrayList<>();
        collectTable.getColumnList().stream()
                .filter(column -> Tool.isNotBlank(column.getTargetColumn()))
                .forEach(column -> columnList.add(column.getTargetColumn()));
        return columnList;
    }

    private Map<String, Object> createQueryCountMap(DatasourceBO datasource, CollectTableBO collectTable) {
        Query query = new Query();
        query.setTableName(collectTable.getOriginTable());

        if (CollectTableBO.SYNC_TYPE_ADD.equals(collectTable.getSyncType())) {
            if (Tool.isBlank(collectTable.getIncrementField())) throw new CommonException("增量字段未配置！");

            if (collectTable.getIncrementStartTime() != null) {
                query.addCondition(new ArrayList<>() {{
                    add(new Condition(collectTable.getIncrementField(), "整数", "大于", collectTable.getIncrementStartTime()));
                }});
            }

            query.addCondition(new ArrayList<>() {{
                add(new Condition(collectTable.getIncrementField(), "整数", "小于等于", collectTable.getIncrementEndTime()));
            }});
        }

        return SqlBuilderFactory.buildQueryMap(datasource.getDatabaseType(), query);
    }

    private Map<String, Object> createQueryListMap(DatasourceBO datasource, CollectTableBO collectTable) {
        Query query = new Query();
        query.setTableName(collectTable.getOriginTable());

        if (CollectTableBO.SYNC_TYPE_ADD.equals(collectTable.getSyncType())) {
            if (Tool.isBlank(collectTable.getIncrementField())) throw new CommonException("增量字段未配置！");
            if (collectTable.getIncrementStartTime() != null) {
                query.addCondition(new ArrayList<>() {{
                    add(new Condition(collectTable.getIncrementField(), "整数", "大于", collectTable.getIncrementStartTime()));
                }});
            }

            query.addCondition(new ArrayList<>() {{
                add(new Condition(collectTable.getIncrementField(), "整数", "小于等于", collectTable.getIncrementEndTime()));
            }});
        }

        if (Tool.isNotBlank(collectTable.getOriginPrimaryKey())) {
            query.addSortField(collectTable.getOriginPrimaryKey(), "asc", 1);
        } else if (Tool.isNotBlank(collectTable.getIncrementField())) {
            query.addSortField(collectTable.getIncrementField(), "asc", 1);
        }

        return SqlBuilderFactory.buildQueryMap(datasource.getDatabaseType(), query);
    }

    private void convertDataList(CollectTableBO collectTable, List<Map<String, Object>> dataList) {
        if (collectTable.getColumnList().isEmpty() || dataList.isEmpty()) return;

        for (Map<String, Object> data : dataList) {
            for (CollectColumnBO column : collectTable.getColumnList()) {
                Object origin = Tool.isBlank(column.getOriginColumn()) ? null : data.get(column.getOriginColumn());
                Object target = DataConvertFactory.convert(column.getDataConvert(), column.getConvertParam(), origin);
                data.put(column.getTargetColumn(), target);
            }
        }
    }

    private void saveDataList(
            DatasourceBO targetDatasource,
            CollectTableBO collectTable,
            CollectTableLogBO collectTableLog,
            List<Map<String, Object>> dataList,
            List<String> columnList) {
        if (dataList.isEmpty()) return;

        if (CollectTableBO.WRITE_MODE_DW.equals(collectTable.getWriteMode())) {
            insertDataList(targetDatasource, collectTable, collectTableLog, dataList, columnList);
        } else {
            if (dataList.size() > 1000) {
                List<Map<String, Object>> resultList = new ArrayList<>();
                for (Map<String, Object> data : dataList) {
                    resultList.add(data);

                    if (resultList.size() >= 1000) {
                        updateDataList(targetDatasource, collectTable, collectTableLog, resultList, columnList);
                        resultList = new ArrayList<>();
                    }
                }
                if (!resultList.isEmpty()) {
                    updateDataList(targetDatasource, collectTable, collectTableLog, resultList, columnList);
                }
            } else {
                updateDataList(targetDatasource, collectTable, collectTableLog, dataList, columnList);
            }
        }
    }

    private void insertDataList(
            DatasourceBO datasource,
            CollectTableBO collectTable,
            CollectTableLogBO collectTableLog,
            List<Map<String, Object>> dataList,
            List<String> columnList) {
        if (dataList.isEmpty()) return;

        addCount += dataList.size();
        DatasourceUtil.update(datasource, collectTable.getTargetTable(), null, dataList, columnList);
        collectTableLog.setAddCount(collectTableLog.getAddCount() + dataList.size());
    }

    private void updateDataList(
            DatasourceBO datasource,
            CollectTableBO collectTable,
            CollectTableLogBO collectTableLog,
            List<Map<String, Object>> dataList,
            List<String> columnList) {
        if (dataList.isEmpty()) return;

        if (Tool.isBlank(collectTable.getOriginPrimaryKey())) throw new CommonException("来源表主键字段未配置！");
        if (Tool.isBlank(collectTable.getTargetPrimaryKey())) throw new CommonException("来源表主键字段未配置！");

        String ids = dataList.stream().map(item -> String.valueOf(item.get(collectTable.getOriginPrimaryKey())))
                .collect(Collectors.joining(","));
        if (Tool.isBlank(ids)) throw new CommonException("来源数据主键字段值为空！");

        // 按主键批量查询存在的数据
        Query query = new Query();
        query.setTableName(collectTable.getTargetTable());
        query.addQueryField(new QueryField(collectTable.getTargetPrimaryKey(), collectTable.getTargetPrimaryKey()));
        query.addCondition(new ArrayList<>() {{
            add(new Condition(collectTable.getTargetPrimaryKey(), "字符", "in", ids));
        }});
        Map<String, Object> paramMap = SqlBuilderFactory.buildQueryMap(datasource.getDatabaseType(), query);
        List<Map<String, Object>> queryList = DatasourceUtil.queryList(datasource, paramMap);
        if (queryList.isEmpty()) { // 主键不存在，全为新增
            addCount += dataList.size();
            DatasourceUtil.update(datasource, collectTable.getTargetTable(), null, dataList, columnList);
            collectTableLog.setAddCount(collectTableLog.getAddCount() + dataList.size());
        } else if (queryList.size() == dataList.size()) { // 全更新
            updateCount += dataList.size();
            DatasourceUtil.update(
                    datasource,
                    collectTable.getTargetTable(),
                    collectTable.getTargetPrimaryKey(),
                    dataList,
                    columnList);
            collectTableLog.setUpdateCount(collectTableLog.getUpdateCount() + dataList.size());
        } else {
            List<Map<String, Object>> addList = new ArrayList<>();
            List<Map<String, Object>> updateList = new ArrayList<>();

            for (Map<String, Object> data : dataList) {
                if (queryList.stream().anyMatch(item -> item.get(collectTable.getTargetPrimaryKey())
                        .equals(data.get(collectTable.getOriginPrimaryKey())))) {
                    updateList.add(data);
                } else {
                    addList.add(data);
                }
            }

            DatasourceUtil.update(datasource, collectTable.getTargetTable(), null, addList, columnList);
            DatasourceUtil.update(
                    datasource,
                    collectTable.getTargetTable(),
                    collectTable.getTargetPrimaryKey(),
                    updateList,
                    columnList);

            collectTableLog.setAddCount(collectTableLog.getAddCount() + addList.size());
            collectTableLog.setUpdateCount(collectTableLog.getUpdateCount() + updateList.size());
        }
    }

    public int getAddCount() {
        return addCount;
    }

    public int getUpdateCount() {
        return updateCount;
    }

    public void shutdown() {
        shutdown = true;
    }

    public boolean isSuccess() {
        return success;
    }

    private CollectTableLogBO createLog(String taskLogId, CollectTableBO collectTable) {
        CollectTableLogBO collectTableLog = new CollectTableLogBO();
        collectTableLog.setId(UUID.getUUID());
        collectTableLog.setTaskLogId(taskLogId);
        collectTableLog.setCollectTableId(collectTable.getId());
        collectTableLog.setOriginTable(collectTable.getOriginTable());
        collectTableLog.setTargetTable(collectTable.getTargetTable());
        collectTableLog.setStartTime(System.currentTimeMillis());
        collectTableLog.setExecuteStatus("执行中");
        collectTableLog.setAddCount(0L);
        collectTableLog.setUpdateCount(0L);
        return collectTableLog;
    }
}
